package com.orion.service;

import cn.hutool.json.JSONUtil;
import com.orion.entity.User;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author Administrator
 * @date 2021/12/2
 */
@Component
public class MessageConsumerService {

    @Resource
    private UserService userService;

    @Resource
    private KafkaTemplate kafkaTemplate;

    /**
     * 设置consumer的auto_offset_set为earliest，
     * b项目设置latest，测试latest不会消费已存在的offset只会消息最新生产的消息，而earliest则会消费
     * <p>
     * <p>
     * 和b项目监听同一个topic，而且是同一个group，即等于集群，默认concurrcy是3，先后启动ab，可见重平衡的情况
     *
     * @param records
     * @param ack
     */
    /*@KafkaListener(id = "haha", groupId = "sadan", topics = {"llc2021","llc2022"}, concurrency = "2")
    public void singleConsume01(List<ConsumerRecord<String, Object>> records, Acknowledgment ack) {
        System.out.println("--------- 接收并处理消息 ---------");
        for (ConsumerRecord<String, Object> record : records) {
            System.out.println(Thread.currentThread() + "-----消费者收到消息:" + record + "; ack:" + ack);
        }
        ack.acknowledge();
    }*/

    /**
     * 设置consumer的auto_offset_set为earliest，
     * b项目设置latest，测试latest不会消费已存在的offset只会消息最新生产的消息，而earliest则会消费
     * <p>
     * <p>
     * 和b项目监听同一个topic，而且是同一个group，即等于集群，默认concurrcy是3，先后启动ab，可见重平衡的情况
     *
     * @param records
     */
    @KafkaListener(id = "haha", groupId = "sadan", topics = {"llc2021", "llc2022"}, clientIdPrefix = "orion", concurrency = "2"
            , errorHandler = "listenSeekingErrorHandler")
    public void singleConsume01(@Header(KafkaHeaders.RAW_DATA) List<ConsumerRecord<String, Object>> records,
                                @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
                                @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                                @Header(KafkaHeaders.OFFSET) List<Long> offsets,
                                @Header(KafkaHeaders.CONSUMER) Consumer consumer) {
        System.out.println("--------- 接收并处理消息 ---------" + topics + "--" + partitions + " --" + offsets);
        for (ConsumerRecord<String, Object> record : records) {
            System.out.println(Thread.currentThread() + "-----消费者收到消息:" + record);
            consumer.commitSync();
        }
        //ack.acknowledge();
    }

    // 消费监听
    @KafkaListener(topics = {"topic3"}, groupId = "single")
    public void listen2(List<ConsumerRecord<String, Object>> records,
                        Acknowledgment ack) {
        System.out.println("--------- 接收并处理消息 ---------");
        try {
            for (ConsumerRecord<String, Object> record : records) {
                System.out.println(Thread.currentThread() + "-----消费者收到消息:" + record);
                if ("micro5512".equals(record.key())) {
                    throw new RuntimeException("rt ex");
                }
            }
        } finally {
            ack.acknowledge();
        }
    }

    //死信处理
    @KafkaListener(topics = {"topic3.DLT"}, groupId = "single")
    public void listen2_DLT(List<ConsumerRecord<String, Object>> records,
                            Acknowledgment ack) {
        System.out.println("--------- DLT接收并处理消息 ---------");
        try {
            for (ConsumerRecord<String, Object> record : records) {
                System.out.println(Thread.currentThread() + "-----消费者收到消息:" + record);
                if ("micro5512".equals(record.key())) {
                    throw new RuntimeException("dlt rt ex");
                }
            }
            ack.acknowledge();
        } catch (Exception e) {
            System.out.println("dlt 还是失败，后续处理");
        }
    }

    // 消费监听
    @KafkaListener(topics = {"db"}, groupId = "single")
    public void dbTest(List<ConsumerRecord<String, Object>> records, Acknowledgment ack) {
        System.out.println("--------- db接收并处理消息 ---------");
        try {
            for (ConsumerRecord<String, Object> record : records) {
                System.out.println(Thread.currentThread() + "-----db消费者收到消息:" + record + "--" + ack);
                User user = JSONUtil.toBean(record.value().toString(),User.class);
                System.out.println("--db user:" + user);
                boolean result = userService.save(user);
                if (!result){
                    //往上抛，后续的message都不会处理
                    //throw new RuntimeException("rpc error");
                    ListenableFuture future = kafkaTemplate.send(record.topic() + "DLT", record.key(), record.value());
                    future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                        @Override
                        public void onFailure(Throwable ex) {
                            System.out.println("转发到死信队列失败:" + ex);
                        }

                        @Override
                        public void onSuccess(SendResult<String, Object> result) {
                            System.out.println("转发到死信队列成功:" + result.toString());
                        }
                    });
                }
            }
        } finally {
            ack.acknowledge();
        }
    }

    //死信处理
    @KafkaListener(topics = {"db.DLT"}, groupId = "single")
    public void dbTest_DLT(List<ConsumerRecord<String, Object>> records, Acknowledgment ack) {
        System.out.println("--------- db.DLT接收并处理消息 ---------");
        try {
            for (ConsumerRecord<String, Object> record : records) {
                System.out.println(Thread.currentThread() + "-----db.DLT消费者收到消息:" + record + "--" + ack);

                    throw new RuntimeException("dlt rt ex");

            }
            ack.acknowledge();
        } catch (Exception e) {
            System.out.println("dlt 还是失败，后续处理");
        }
    }
}
